package com.ndp.fb.rdb.mongo.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSON;
import com.mongodb.AggregationOptions;
import com.mongodb.BasicDBObject;
import com.mongodb.Bytes;
import com.mongodb.Cursor;
import com.mongodb.DB;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.ndp.fb.constant.MongoConstants;
import com.ndp.fb.model.AdSetStats;
import com.ndp.fb.mongo.api.AdSetStatsHistoryMongoService;
import com.ndp.fb.util.DateFormatUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by jacky.cheng on 2015/6/29.
 */
@Service
public class AdSetStatsHistoryMongoServiceImpl implements AdSetStatsHistoryMongoService {
    Logger logger = LoggerFactory.getLogger(getClass());
    @Autowired
    private DB adHistoryDBClient;

    @Override
    public List<AdSetStats> getAllSnap(Long adSetId, Date sinceDay, Date untilDay, String... fields) {
        List<AdSetStats> result = new ArrayList<>();
        if (adSetId == null) {
            logger.warn("adSetId is null. return.");
            return result;
        }
        // 查询条件
        DBObject condition = new BasicDBObject();
        condition.put("campaignId", adSetId);
        DBObject timeObject = new BasicDBObject();
        timeObject.put("$gte", sinceDay);
        timeObject.put("$lte", untilDay);
        condition.put("updateTimeList.0", timeObject);
        DBObject dayObject = new BasicDBObject();
        dayObject.put("$gte", DateFormatUtil.formatLong10(sinceDay));
        dayObject.put("$lte", DateFormatUtil.formatLong10(untilDay));
        condition.put("day", dayObject);
        //返回字段
        DBObject keys = new BasicDBObject();
        for (String queryField : fields) {
            keys.put(queryField, 1);
        }
        DBCursor dataCur = adHistoryDBClient.getCollection(MongoConstants.TABLE_NAME_ADSET_STATS_BY_DAY_HISTORY).find(condition, keys).addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        try {
            if (dataCur == null) {
                logger.warn("没有查到统计快照数据。adSetId={}", adSetId);
                return null;
            }
            while (dataCur.hasNext()) {
                DBObject next = dataCur.next();
                if (next != null) {
                    AdSetStats adSetStats = JSON.parseObject(JSON.toJSONString(next), AdSetStats.class);
                    List<Date> updateTimeList = (List<Date>) next.get("updateTimeList");
                    adSetStats.setUpdateTime(updateTimeList.get(0));
                    result.add(adSetStats);
                } else {
                    logger.error("getAll,统计数据查询有空结果。");
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            dataCur.close();
        }

        return result;

    }

    @Override
    public Map<Long, Date> findChangedAdSetIds(Date startTime, Date endTime) {
        // 查询条件
        DBObject matchChildObject = new BasicDBObject();
        // updateTimeList是每条统计数据的更新时间历史，第0个元素就代表了这条记录的创建时间，如果创建时间大于给定的查询开始时间则说明该记录是新的，也就表明了该AdSet的统计数据发生了变化
        matchChildObject.put("$gt", startTime);
        matchChildObject.put("$lte", endTime);
        DBObject projectChildDBObject = new BasicDBObject();
        projectChildDBObject.put("campaignId", 1);
        projectChildDBObject.put("updateTimeList", 1);
        List<DBObject> aggregateObjects = new ArrayList<>();
        aggregateObjects.add(new BasicDBObject("$match", new BasicDBObject("updateTimeList.0", matchChildObject)));
        aggregateObjects.add(new BasicDBObject("$project", projectChildDBObject));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = adHistoryDBClient.getCollection(MongoConstants.TABLE_NAME_ADSET_STATS_BY_DAY_HISTORY).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        Map<Long, Date> results = new HashMap<>();
        try {
            if (cursor == null) {
                logger.warn("没有查到变化的adset数据");
                return null;
            }
            while (cursor.hasNext()) {
                DBObject next = cursor.next();
                Long campaignId = (Long) next.get("campaignId");
                Date timeStamp = (Date) ((List) next.get("updateTimeList")).get(0);
                Date existTimeStamp = results.get(campaignId);
                if (null == existTimeStamp || (null != existTimeStamp && timeStamp.after(existTimeStamp))) {    //如果是第一次放入或者新的时间戳大于已经存进去的时间戳，则覆盖时间戳
                    results.put(campaignId, timeStamp);
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }

        return results;
    }
}
